/*-------------------------------------------------------------------------
 *
 * blinsert.c
 *		Bloom 索引构建和插入函数。
 *
 * Copyright (c) 2016-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  contrib/bloom/blinsert.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/genam.h"
#include "access/generic_xlog.h"
#include "access/tableam.h"
#include "bloom.h"
#include "catalog/index.h"
#include "miscadmin.h"
#include "storage/bufmgr.h"
#include "storage/indexfsm.h"
#include "storage/smgr.h"
#include "utils/memutils.h"
#include "utils/rel.h"

PG_MODULE_MAGIC;

/*
 * Bloom索引构建状态。我们在这里累积一页数据，然后
 * 将其刷新到缓冲区管理器。
 */
typedef struct
{
	BloomState	blstate;		/* bloom索引状态 */
	int64		indtuples;		/* 已索引的元组总数 */
	MemoryContext tmpCtx;		/* 临时内存上下文在每个
								 * 元组后重置 */
	PGAlignedBlock data;		/* 缓存页面 */
	int			count;			/* 缓存页面中的元组数量 */
} BloomBuildState;

/*
 * 刷新BloomBuildState中缓存的页面。
 */
static void fc_flushCachedPage(Relation fc_index, BloomBuildState *fc_buildstate)
{
	Page		fc_page;
	Buffer		fc_buffer = BloomNewBuffer(fc_index);
	GenericXLogState *fc_state;

	fc_state = GenericXLogStart(fc_index);
	fc_page = GenericXLogRegisterBuffer(fc_state, fc_buffer, GENERIC_XLOG_FULL_IMAGE);
	memcpy(fc_page, fc_buildstate->data.data, BLCKSZ);
	GenericXLogFinish(fc_state);
	UnlockReleaseBuffer(fc_buffer);
}

/*
 * （重新）初始化BloomBuildState中的缓存页面。
 */
static void fc_initCachedPage(BloomBuildState *fc_buildstate)
{
	BloomInitPage(fc_buildstate->data.data, 0);
	fc_buildstate->count = 0;
}

/*
 * 每个元组的回调，用于table_index_build_scan。
 */
static void fc_bloomBuildCallback(Relation fc_index, ItemPointer fc_tid, Datum *fc_values,
				   bool *fc_isnull, bool fc_tupleIsAlive, void *fc_state)
{
	BloomBuildState *fc_buildstate = (BloomBuildState *) fc_state;
	MemoryContext fc_oldCtx;
	BloomTuple *fc_itup;

	fc_oldCtx = MemoryContextSwitchTo(fc_buildstate->tmpCtx);

	fc_itup = BloomFormTuple(&fc_buildstate->blstate, fc_tid, fc_values, fc_isnull);

	/* 尝试将下一个项添加到缓存页面 */
	if (BloomPageAddItem(&fc_buildstate->blstate, fc_buildstate->data.data, fc_itup))
	{
		/* 下一个项已成功添加 */
		fc_buildstate->count++;
	}
	else
	{
		/* 缓存页面已满，刷新它并创建新页面 */
		fc_flushCachedPage(fc_index, fc_buildstate);

		CHECK_FOR_INTERRUPTS();

		fc_initCachedPage(fc_buildstate);

		if (!BloomPageAddItem(&fc_buildstate->blstate, fc_buildstate->data.data, fc_itup))
		{
			/* 我们不应该在这里，因为我们正在插入到空页面 */
			elog(ERROR, "could not add new bloom tuple to empty page");
		}

		/* 下一个项已成功添加 */
		fc_buildstate->count++;
	}

	/* 更新总元组计数 */
	fc_buildstate->indtuples += 1;

	MemoryContextSwitchTo(fc_oldCtx);
	MemoryContextReset(fc_buildstate->tmpCtx);
}

/*
 * 构建新的bloom索引。
 */
IndexBuildResult * blbuild(Relation fc_heap, Relation fc_index, IndexInfo *fc_indexInfo)
{
	IndexBuildResult *fc_result;
	double		fc_reltuples;
	BloomBuildState fc_buildstate;

	if (RelationGetNumberOfBlocks(fc_index) != 0)
		elog(ERROR, "index \"%s\" already contains data",
			 RelationGetRelationName(fc_index));

	/* 初始化元页面 */
	BloomInitMetapage(fc_index);

	/* 初始化bloom构建状态 */
	memset(&fc_buildstate, 0, sizeof(fc_buildstate));
	initBloomState(&fc_buildstate.blstate, fc_index);
	fc_buildstate.tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
											  "Bloom build temporary context",
											  ALLOCSET_DEFAULT_SIZES);
	fc_initCachedPage(&fc_buildstate);

	/* 执行堆扫描 */
	fc_reltuples = table_index_build_scan(fc_heap, fc_index, fc_indexInfo, true, true,
									   fc_bloomBuildCallback, (void *) &fc_buildstate,
									   NULL);

	/* 如果需要则刷新最后一个页面（除非堆为空） */
	if (fc_buildstate.count > 0)
		fc_flushCachedPage(fc_index, &fc_buildstate);

	MemoryContextDelete(fc_buildstate.tmpCtx);

	fc_result = (IndexBuildResult *) palloc(sizeof(IndexBuildResult));
	fc_result->heap_tuples = fc_reltuples;
	fc_result->index_tuples = fc_buildstate.indtuples;

	return fc_result;
}

/*
 * 在初始化分支中构建一个空的bloom索引。
 */
void blbuildempty(Relation fc_index)
{
	Page		fc_metapage;

	/* 构造元数据页。 */
	fc_metapage = (Page) palloc(BLCKSZ);
	BloomFillMetapage(fc_index, fc_metapage);

	/*
	 * 写入页面并记录。看似立即同步应该足以保证文件
	 * 在磁盘上存在，但恢复本身可能会在重放时将其删除，例如
	 * XLOG_DBASE_CREATE* 或 XLOG_TBLSPC_CREATE记录。因此，
	 * 即使wal_level=minimal，我们也需要这样做。
	 */
	PageSetChecksumInplace(fc_metapage, BLOOM_METAPAGE_BLKNO);
	smgrwrite(RelationGetSmgr(fc_index), INIT_FORKNUM, BLOOM_METAPAGE_BLKNO,
			  (char *) fc_metapage, true);
	log_newpage(&(RelationGetSmgr(fc_index))->smgr_rnode.node, INIT_FORKNUM,
				BLOOM_METAPAGE_BLKNO, fc_metapage, true);

	/*
	 * 即使我们记录了页面，立即同步也是必要的，因为
	 * 写入没有经过shared_buffers，因此一个并发
	 * 检查点可能已经将重做指针移过我们的xlog记录。
	 */
	smgrimmedsync(RelationGetSmgr(fc_index), INIT_FORKNUM);
}

/*
 * 将新元组插入到bloom索引中。
 */
bool blinsert(Relation fc_index, Datum *fc_values, bool *fc_isnull,
		 ItemPointer fc_ht_ctid, Relation fc_heapRel,
		 IndexUniqueCheck fc_checkUnique,
		 bool fc_indexUnchanged,
		 IndexInfo *fc_indexInfo)
{
	BloomState	fc_blstate;
	BloomTuple *fc_itup;
	MemoryContext fc_oldCtx;
	MemoryContext fc_insertCtx;
	BloomMetaPageData *fc_metaData;
	Buffer		fc_buffer,
				fc_metaBuffer;
	Page		fc_page,
				fc_metaPage;
	BlockNumber fc_blkno = InvalidBlockNumber;
	OffsetNumber fc_nStart;
	GenericXLogState *fc_state;

	fc_insertCtx = AllocSetContextCreate(CurrentMemoryContext,
									  "Bloom insert temporary context",
									  ALLOCSET_DEFAULT_SIZES);

	fc_oldCtx = MemoryContextSwitchTo(fc_insertCtx);

	initBloomState(&fc_blstate, fc_index);
	fc_itup = BloomFormTuple(&fc_blstate, fc_ht_ctid, fc_values, fc_isnull);

	/*
	 * 起初，尝试将新元组插入到notFullPage数组中的第一页面。
	 * 如果成功，我们不需要修改元页面。
	 */
	fc_metaBuffer = ReadBuffer(fc_index, BLOOM_METAPAGE_BLKNO);
	LockBuffer(fc_metaBuffer, BUFFER_LOCK_SHARE);
	fc_metaData = BloomPageGetMeta(BufferGetPage(fc_metaBuffer));

	if (fc_metaData->nEnd > fc_metaData->nStart)
	{
		Page		fc_page;

		fc_blkno = fc_metaData->notFullPage[fc_metaData->nStart];
		Assert(fc_blkno != InvalidBlockNumber);

		/* 插入时不要持有metabuffer锁 */
		LockBuffer(fc_metaBuffer, BUFFER_LOCK_UNLOCK);

		fc_buffer = ReadBuffer(fc_index, fc_blkno);
		LockBuffer(fc_buffer, BUFFER_LOCK_EXCLUSIVE);

		fc_state = GenericXLogStart(fc_index);
		fc_page = GenericXLogRegisterBuffer(fc_state, fc_buffer, 0);

		/*
		 * 我们可能找到了一个最近被VACUUM删除的页面。
		 * 如果是这样，我们可以重用它，但必须重新初始化。
		 */
		if (PageIsNew(fc_page) || BloomPageIsDeleted(fc_page))
			BloomInitPage(fc_page, 0);

		if (BloomPageAddItem(&fc_blstate, fc_page, fc_itup))
		{
			/* 成功！应用更改，清理并退出 */
			GenericXLogFinish(fc_state);
			UnlockReleaseBuffer(fc_buffer);
			ReleaseBuffer(fc_metaBuffer);
			MemoryContextSwitchTo(fc_oldCtx);
			MemoryContextDelete(fc_insertCtx);
			return false;
		}

		/* 没有适合，必须尝试其他页面 */
		GenericXLogAbort(fc_state);
		UnlockReleaseBuffer(fc_buffer);
	}
	else
	{
		/* notFullPage中没有条目 */
		LockBuffer(fc_metaBuffer, BUFFER_LOCK_UNLOCK);
	}

	/*
	 * 尝试notFullPage数组中的其他页面。我们必须更改
	 * 元页面中的nStart。因此，获取元页面的独占锁。
	 */
	LockBuffer(fc_metaBuffer, BUFFER_LOCK_EXCLUSIVE);

	/* nStart在我们没有锁定时可能已更改 */
	fc_nStart = fc_metaData->nStart;

	/* 如果我们已经在上面尝试过，则跳过第一页 */
	if (fc_nStart < fc_metaData->nEnd &&
		fc_blkno == fc_metaData->notFullPage[fc_nStart])
		fc_nStart++;

	/*
	 * 此循环迭代我们从notFullPage数组中尝试的每个页面，并
	 * 将为分配新页面的后备情况初始化一个GenericXLogState。
	 */
	for (;;)
	{
		fc_state = GenericXLogStart(fc_index);

		/* 获取元页面的可修改副本 */
		fc_metaPage = GenericXLogRegisterBuffer(fc_state, fc_metaBuffer, 0);
		fc_metaData = BloomPageGetMeta(fc_metaPage);

		if (fc_nStart >= fc_metaData->nEnd)
			break;				/* notFullPage数组中没有更多条目 */

		fc_blkno = fc_metaData->notFullPage[fc_nStart];
		Assert(fc_blkno != InvalidBlockNumber);

		fc_buffer = ReadBuffer(fc_index, fc_blkno);
		LockBuffer(fc_buffer, BUFFER_LOCK_EXCLUSIVE);
		fc_page = GenericXLogRegisterBuffer(fc_state, fc_buffer, 0);

		/* 基本上与上述逻辑相同 */
		if (PageIsNew(fc_page) || BloomPageIsDeleted(fc_page))
			BloomInitPage(fc_page, 0);

		if (BloomPageAddItem(&fc_blstate, fc_page, fc_itup))
		{
			/* 成功！应用更改，清理并退出 */
			fc_metaData->nStart = fc_nStart;
			GenericXLogFinish(fc_state);
			UnlockReleaseBuffer(fc_buffer);
			UnlockReleaseBuffer(fc_metaBuffer);
			MemoryContextSwitchTo(fc_oldCtx);
			MemoryContextDelete(fc_insertCtx);
			return false;
		}

		/* 没有适合，必须尝试其他页面 */
		GenericXLogAbort(fc_state);
		UnlockReleaseBuffer(fc_buffer);
		fc_nStart++;
	}

	/*
	 * 在 notFullPage 数组中没有找到插入的位置。 分配新页面。
	 * (XXX 在持有元页面的 ex-lock 时这样做好吗？？)
	 */
	fc_buffer = BloomNewBuffer(fc_index);

	fc_page = GenericXLogRegisterBuffer(fc_state, fc_buffer, GENERIC_XLOG_FULL_IMAGE);
	BloomInitPage(fc_page, 0);

	if (!BloomPageAddItem(&fc_blstate, fc_page, fc_itup))
	{
		/* 我们不应该在这里，因为我们正在插入到一个空页面 */
		elog(ERROR, "could not add new bloom tuple to empty page");
	}

	/* 重置 notFullPage 数组以仅包含这个新页面 */
	fc_metaData->nStart = 0;
	fc_metaData->nEnd = 1;
	fc_metaData->notFullPage[0] = BufferGetBlockNumber(fc_buffer);

	/* 应用更改，清理并退出 */
	GenericXLogFinish(fc_state);

	UnlockReleaseBuffer(fc_buffer);
	UnlockReleaseBuffer(fc_metaBuffer);

	MemoryContextSwitchTo(fc_oldCtx);
	MemoryContextDelete(fc_insertCtx);

	return false;
}
